/* 异步工作器 */

#ifndef __M_LOOPER_H
#define __M_LOOPER_H

#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>

namespace log{
    using Functor = std::function<void(Buffer&)>;
    enum class AsyncType{
        ASYNC_SAVE, //安全状态，表示缓冲区满了则阻塞，避免资源消耗
        ASYNC_UNSAVE//不考虑资源耗尽的问题，无限扩容，常用于测试  
    };

    class AsyncLooper{
        public:
            using ptr = std::shared_ptr<AsyncLooper>;
            AsyncLooper(const Functor& cb, AsyncType loop_type = AsyncType::ASYNC_SAVE)
            :_stop(false), _thread(std::thread(&AsyncLooper::threadEntry, this))
            ,_callback(cb), _lopper_type(loop_type)
            {}  
            ~AsyncLooper(){
                stop();
            }
            void stop(){
                _stop = true;
                _cond_con.notify_all(); //唤醒工作线程
                 _thread.join();
            }
            void push(const char* data, size_t len){
                //1.无限扩容 -- 非安全   2.固定大小 -- 生产缓冲区中数据满了就阻塞
                std::unique_lock<std::mutex> lock(_mutex);
                //条件变量空值，若缓冲区中剩余空间大小大于数据长度，则可以进行添加数据
                if(_lopper_type == AsyncType::ASYNC_SAVE)
                    _cond_pro.wait(lock, [&](){return _pro_buf.writerAbleSize() >= len; });
                //能走下来说明满足了条件，可以向缓冲区中添加数据
                _pro_buf.push(data, len);
                //唤醒消费者对缓冲区的处理
                _cond_con.notify_one();
            }
        private:
            //线程入口函数 -- 对消费缓冲区的数据进行处理
            void threadEntry(){
                while(1){
                    //  为互斥锁设置一个生命周期，当缓冲区交换完毕后就解锁（并不对数据的处理过程进行加锁保护，否则变成串行）
                    {
                        //1.判断生产缓存区有没有数据，有则交换，无则阻塞
                        std::unique_lock<std::mutex> lock(_mutex);
                        //退出标志被设置，且生产缓冲区中已无数据，这时候再退出，否则有可能生产缓冲区中仍然有数据，但是没有被处理
                        if(_stop && _pro_buf.empty()) { break;}
                        //若当前是退出前被唤醒，或者有数据被唤醒，则返回真，继续向下运行，否则继续休眠
                        _cond_con.wait(lock, [&](){return _stop || !_pro_buf.empty(); });   
                        _con_buf.swap(_pro_buf);
                        //2.唤醒生产者
                        if(_lopper_type == AsyncType::ASYNC_SAVE)
                            _cond_pro.notify_all();
                    }
                    //3.被唤醒后，对消费缓冲区进行处理
                    _callback(_con_buf);
                    //4.初始化消费缓存区
                    _con_buf.reset();
                }
            }
        private:
            Functor _callback;//具体对缓冲区数据进行处理的回调函数，由异步工作器使用者传入

        private:
            AsyncType _lopper_type;
            std::atomic<bool> _stop;//工作器停止标志
            Buffer _pro_buf; //生产缓冲区
            Buffer _con_buf; //消费缓冲区
            std::mutex _mutex;
            std::condition_variable _cond_pro;
            std::condition_variable _cond_con;
            std::thread _thread;//异步工作器对于工作线程
    };
}

#endif